
package simpledb;

import java.util.*;

/**
 * Knows how to compute some aggregate over a set of IntFields.
 */
public class IntegerAggregator implements Aggregator {

    private static final long serialVersionUID = 1L;

    private int gbfield;
    private Type gbfieldtype;
    private int afield;
    private Op what;

    private Map<Field,Integer> groupMap;
    private Map<Field,Integer> countMap;
    private Map<Field,List<Integer>> avgMap;

    /**
     * Aggregate constructor
     *
     * @param gbfield
     *            the 0-based index of the group-by field in the tuple, or
     *            NO_GROUPING if there is no grouping
     * @param gbfieldtype
     *            the type of the group by field (e.g., Type.INT_TYPE), or null
     *            if there is no grouping
     * @param afield
     *            the 0-based index of the aggregate field in the tuple
     * @param what
     *            the aggregation operator
     */

    public IntegerAggregator(int gbfield, Type gbfieldtype, int afield, Op what) {
        // some code goes here
        this.gbfield = gbfield;
        this.gbfieldtype = gbfieldtype;
        this.afield = afield;
        this.what = what;
        this.groupMap = new HashMap<>();
        this.avgMap = new HashMap<>();
        this.countMap = new HashMap<>();
    }

    /**
     * Merge a new tuple into the aggregate, grouping as indicated in the
     * constructor
     *
     * @param tup
     *            the Tuple containing an aggregate field and a group-by field
     */
    public void mergeTupleIntoGroup(Tuple tup) {
        // some code goes here
        //获取afield和gbfield对应的值
        IntField afield = (IntField) tup.getField(this.afield);
        int temp = afield.getValue();
        Field gbfield;
        if(this.gbfield!=NO_GROUPING){
            gbfield=tup.getField(this.gbfield);
        }else{
            gbfield=null;
        }
        //进行聚合操作
        switch(this.what){
            case MIN:
                if(!groupMap.containsKey(gbfield)){
                    groupMap.put(gbfield,temp);
                }else{
                    groupMap.put(gbfield,Math.min(groupMap.get(gbfield),temp));
                }
                break;
            case MAX:
                if(!groupMap.containsKey(gbfield)){
                    groupMap.put(gbfield,temp);
                }else{
                    groupMap.put(gbfield,Math.max(groupMap.get(gbfield),temp));
                }
                break;
            case SUM:
                if(!groupMap.containsKey(gbfield)){
                    groupMap.put(gbfield,temp);
                }else{
                    groupMap.put(gbfield,groupMap.get(gbfield)+temp);
                }
                break;
            case COUNT:
                if(!groupMap.containsKey(gbfield)){
                    groupMap.put(gbfield,1);
                }else{
                    groupMap.put(gbfield,groupMap.get(gbfield)+1);
                }
                break;
            case SC_AVG://没太弄明白这两个的含义

            case SUM_COUNT:

            case AVG:
                if(!avgMap.containsKey(gbfield)){
                    List<Integer> values = new ArrayList<Integer>();
                    values.add(temp);
                    avgMap.put(gbfield,values);
                }else{
                    List<Integer> values = avgMap.get(gbfield);
                    values.add(temp);
                }
                break;
            default:
                throw new IllegalArgumentException("Aggregate not supported!");
        }
    }

    /**
     * Create a OpIterator over group aggregate results.
     *
     * @return a OpIterator whose tuples are the pair (groupVal, aggregateVal)
     *         if using group, or a single (aggregateVal) if no grouping. The
     *         aggregateVal is determined by the type of aggregate specified in
     *         the constructor.
     */
    public OpIterator iterator() {
        // some code goes here
        return new IntegerAggregatorIterator();
    }

    private class IntegerAggregatorIterator extends AggregateIterator {

        //主要对于求平均的聚合操作进行迭代，其他操作用父类迭代器即可
        private Iterator<Map.Entry<Field, List<Integer>>> avgIt;//avgGroup的迭代器
        private boolean isAvg;

        IntegerAggregatorIterator() {
            super(groupMap, gbfieldtype);
            this.isAvg = what.equals(Op.AVG);
        }

        @Override
        public void open() throws DbException, TransactionAbortedException {
            super.open();
            if (this.isAvg)
                this.avgIt = avgMap.entrySet().iterator();
            else {
                this.avgIt = null;
            }
        }

        @Override
        public boolean hasNext() throws DbException, TransactionAbortedException {
            if (this.isAvg)
                return avgIt.hasNext();
            return super.hasNext();
        }

        @Override
        public Tuple next() throws DbException, TransactionAbortedException, NoSuchElementException {
            Tuple rtn = new Tuple(td);
            if (this.isAvg) {
                Map.Entry<Field, List<Integer>> avgEntry = this.avgIt.next();
                Field avgField = avgEntry.getKey();
                List<Integer> avgList = avgEntry.getValue();
                int value = this.sumList(avgList) / avgList.size();
                this.setFields(rtn, value, avgField);
                return rtn;
            }
            return super.next();
        }

        @Override
        public void rewind() throws DbException, TransactionAbortedException {
            super.rewind();
            if (this.isAvg )
                this.avgIt = avgMap.entrySet().iterator();
        }

        @Override
        public TupleDesc getTupleDesc() {
            return super.getTupleDesc();
        }

        @Override
        public void close() {
            super.close();
            this.avgIt = null;
        }

        private int sumList(List<Integer> l) {
            int sum = 0;
            for (int i : l)
                sum += i;
            return sum;
        }
    }

}